package com.boot.study.wc

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._

/**
 * 流处理统计
 */
object StreamWordCout {

  def main(args: Array[String]): Unit = {
    // 创建批处理执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行
    // env.setParallelism(8)

    // 从外部命令中提取参数
    // windows启动 nc64.exe -L -p 9000
    // --host 127.0.0.1 --port 9000
    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    val host: String = parameterTool.get("host")
    val port: Int = parameterTool.getInt("port")
    // 接受socket文本流
    val dataStream: DataStream[String] = env.socketTextStream(host, port)
    // 转换处理分组
    val resultDataStream: DataStream[(String, Int)] = dataStream.flatMap(_.split(" "))
      .filter(_.nonEmpty)
      .map((_, 1))
      .keyBy(0)
      .sum(1)

    resultDataStream.print().setParallelism(1)
    // 启动任务执行
    env.execute("stream word count")
  }
}
